package com.fengkai.filepost.pdfoutstream.test.common.mq.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.concurrent.TimeUnit;

/**
 * @author Fengkai
 * @creat 2021/12/28
 */
public class AsyncProducer {

       public static void main(String[] args) throws Exception {
           DefaultMQProducer pg = new DefaultMQProducer("pg");
           pg.setNamesrvAddr("192.168.157.130:9876");
           pg.setRetryTimesWhenSendAsyncFailed(0);
           //指定新创建的Topic的Queue数量为2 默认为4
           pg.setDefaultTopicQueueNums(2);
           pg.start();

           for (int i = 0; i <100 ; i++) {
               byte[] body = ("HI," + i).getBytes();
               Message message = new Message("AsysncTopic", "asyncTag", body);
               //为消息指定key
               message.setKeys("key-" + i );
               // 异步发送，指定回调
               pg.send(message, new SendCallback() {

                   //当producer接收到MQ 发送来的ACK后就会出发该回调方法执行
                   @Override
                   public void onSuccess (SendResult sendResult) {
                       System.out.println(sendResult);
                   }

                   @Override
                   public void onException (Throwable e) {
                       e.printStackTrace();
                   }
               });
           }
            //异步发送，需要休眠等待返回信息，要是马上关闭就会报错
           TimeUnit.SECONDS.sleep(3);
           pg.shutdown();

          }
}
